package raft

import (
	"context"
	"fmt"
	"time"

	"mit-6.824/src/labgob"
)

const (
	// LOG_DUMP_SPAN  = 200
	LOG_DUMP_SPAN  = 5000
	LOG_DUMP_LIMIT = 32
)

type DebugFunc = func(string, ...interface{})

type LogEntry struct {
	Term    int
	Index   int
	Command interface{}
}

func (l *LogEntry) String() string {
	return fmt.Sprintf("(%d/%d,=%#v)", l.Index, l.Term, l.Command)
}

type LogSlice = []*LogEntry

type LogMeta struct {
	queue []chan LogSlice // the wait list on this item's commit point
}

type LogKeeper struct {
	// configuration
	applyCh chan ApplyMsg

	// managed by main worker
	currentTerm int
	appended    int
	committed   int
	applied     int
	applying    bool // is apply worker running?
	logs        []*LogEntry
	meta        []*LogMeta

	// message passing
	applyIdle        chan bool
	appendCh         chan AppendEvent
	fetchCh          chan FetchEvent
	updateTermCh     chan UpdateTermEvent
	trimCh           chan TrimEvent
	getLogCh         chan GetLogEvent
	getLastLogCh     chan GetLastLogEvent
	commitCh         chan CommitEvent
	getCommitIndexCh chan GetCommitIndexEvent
	searchCh         chan SearchEvent
	saveCh           chan SaveEvent
	loadCh           chan LoadEvent

	// for debugging
	debug DebugFunc
}

func NewLogger(ctx context.Context, majority int, applyCh chan ApplyMsg, debug DebugFunc) *LogKeeper {
	u := &LogKeeper{
		applyCh: applyCh,

		currentTerm: -1,
		appended:    0,
		committed:   0,
		applied:     0,
		applying:    false,
		logs:        make([]*LogEntry, 0),
		meta:        make([]*LogMeta, 0),

		applyIdle:        make(chan bool),
		appendCh:         make(chan AppendEvent),
		fetchCh:          make(chan FetchEvent),
		updateTermCh:     make(chan UpdateTermEvent),
		trimCh:           make(chan TrimEvent),
		getLogCh:         make(chan GetLogEvent),
		getLastLogCh:     make(chan GetLastLogEvent),
		commitCh:         make(chan CommitEvent),
		getCommitIndexCh: make(chan GetCommitIndexEvent),
		searchCh:         make(chan SearchEvent),
		saveCh:           make(chan SaveEvent),
		loadCh:           make(chan LoadEvent),

		debug: debug,
	}

	go u.main(ctx)

	return u
}

// reserve make sure both u.logs and u.meta contain at least n entries.
func (u *LogKeeper) reserve(n int) {
	for n > len(u.logs) {
		u.logs = append(u.logs, nil)
		u.meta = append(u.meta, &LogMeta{})
	}
}

// resetItem resets log entry at index to nil.
func (u *LogKeeper) resetItem(index int) {
	u.logs[index] = nil

	if len(u.meta[index].queue) > 0 {
		u.debug("log: warning: ignores channels in fetch queue. index=%d", index)
	}

	u.meta[index] = &LogMeta{}
}

// resetLogs resets entire logs.
func (u *LogKeeper) resetLogs(logs LogSlice) {
	u.committed = 0
	u.applied = 0
	u.appended = len(logs)
	u.logs = nil
	u.meta = nil
	u.reserve(len(logs))
	copy(u.logs, logs)
}

// append appends a new entry, increasing u.appended by 1.
// it will wake up all goroutines that wait for the new log.
func (u *LogKeeper) append(log *LogEntry) {
	n := u.appended
	u.appended = n + 1

	u.reserve(n + 1)
	u.logs[n] = log

	// notify pending slice queries.
	for _, v := range u.meta[n].queue {
		// v <- u.logs[n : n+1]
		v <- u.getSlice(n, n+1)
		close(v)
	}
	u.meta[n].queue = nil
}

// doCommit marks the log at index as committed directly,
// without checking commit condition.
func (u *LogKeeper) doCommit(ctx context.Context, index int) {
	if index >= u.committed {
		u.debug("log.commit %s", u.logs[index])
		u.committed = index + 1
		u.notifyApply(ctx)
	}
}

// notifyApply returns true if a new apply task has been started.
func (u *LogKeeper) notifyApply(ctx context.Context) bool {
	if !u.applying && u.committed > u.applied {
		u.applying = true
		begin, end := u.applied, u.committed
		u.applied = u.committed

		go func(ctx context.Context) {
			logs := <-u.Fetch(begin, end)

			// since all logs to be applied are committed, Fetch should
			// return them in one request.
			if len(logs) != end-begin {
				panic(fmt.Errorf("expect %d items, got %d", end-begin, len(logs)))
			}

			for _, log := range logs {
				u.debug("log.apply %s", log)
				select {
				case u.applyCh <- ApplyMsg{true, log.Command, log.Index + 1}: // +1: see Raft.replyStart
					u.debug("log.apply %s done", log)
					continue
				case <-ctx.Done():
					return
				}
			}

			select {
			case u.applyIdle <- true:
			case <-ctx.Done():
			}
		}(ctx)

		return true
	} else {
		return false
	}
}

// IMPORTANT: we should return a copy of slice to avoid data races
// between log updates and log references.
func (u *LogKeeper) getSlice(begin, end int) LogSlice {
	s := u.logs[begin:end]
	t := make(LogSlice, end-begin)
	copy(t, s)
	return t
}

func (u *LogKeeper) main(ctx context.Context) {
	ticker := time.NewTicker(LOG_DUMP_SPAN * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-u.applyIdle:
			u.applying = false
			u.notifyApply(ctx)

		case e := <-u.appendCh:
			if e.log.Term == -1 {
				e.log.Term = u.currentTerm
			}

			if e.log.Index == -1 {
				e.log.Index = u.appended
			} else if e.log.Index != u.appended {
				u.debug("log: warning: index does not match: %d vs %d", e.log.Index, u.appended)
			}

			u.debug("log.Append %s", e.log)
			u.append(e.log)
			e.done <- e.log.Index

		case e := <-u.fetchCh:
			if e.begin < u.appended {
				end := u.appended
				if e.end != -1 && e.end < u.appended {
					end = e.end
				}

				// e.done <- u.logs[e.begin:end]
				e.done <- u.getSlice(e.begin, end)
				close(e.done)
			} else {
				if e.options.has(NonBlocking) {
					e.done <- nil
					close(e.done)
				} else {
					u.reserve(e.begin + 1)
					u.meta[e.begin].queue = append(u.meta[e.begin].queue, e.done)
				}
			}

		case e := <-u.updateTermCh:
			if e.term != u.currentTerm {
				u.debug("log.UpdateTerm %d → %d", u.currentTerm, e.term)
				u.currentTerm = e.term
			}
			ack(e.done)

		case e := <-u.trimCh:
			if e.begin < u.committed {
				panic(fmt.Errorf(
					"you're going to trim committed logs! begin=%d, committed=%d",
					e.begin, u.committed))
			}

			if u.appended > e.begin {
				u.debug("log.Trim [%d, +inf)", e.begin)

				for i := e.begin; i < u.appended; i++ {
					u.resetItem(i)
				}

				u.appended = e.begin
			}

			ack(e.done)

		case e := <-u.getLogCh:
			// u.debug("log.GetLog (%d/*)", e.index)
			if 0 <= e.index && e.index < u.appended {
				e.done <- u.logs[e.index]
			} else {
				e.done <- nil
			}

		case e := <-u.getLastLogCh:
			// u.debug("log.GetLastLog")
			if u.appended > 0 {
				e.done <- u.logs[u.appended-1]
			} else {
				e.done <- nil
			}

		case e := <-u.commitCh:
			index := e.index
			if u.appended-1 < index {
				index = u.appended - 1
			}

			// to prevent negative index
			if index >= 0 {
				u.doCommit(ctx, index)
			}

			ack(e.done)

		case e := <-u.getCommitIndexCh:
			e.done <- u.committed - 1

		case e := <-u.searchCh:
			// v := -1
			// for i := 0; i < u.appended; i++ {
			// 	if e.pred(u.logs[i].Term) {
			// 		v = i
			// 		break
			// 	}
			// }

			l, r := -1, u.appended-1
			for l < r {
				m := l + (r-l)/2 + 1

				if e.pred(u.logs[m].Term) {
					l = m
				} else {
					r = m - 1
				}
			}

			e.done <- l

		case e := <-u.saveCh:
			if u.appended < LOG_DUMP_LIMIT {
				u.debug("log.Save: %s", u.logs[:u.appended])
			} else {
				u.debug("log.Save: appended=%d ...%s", u.appended, u.logs[u.appended-LOG_DUMP_LIMIT:u.appended])
			}

			err := e.encoder.Encode(u.logs[:u.appended])
			e.done <- err

		case e := <-u.loadCh:
			u.debug("log.Load")

			var logs LogSlice
			err := e.decoder.Decode(&logs)
			if err == nil {
				u.resetLogs(logs)
				u.debug("logs: %s", u.logs)
			}
			e.done <- err

		// periodically dump all logs.
		case <-ticker.C:
			// to limit output size
			if u.appended < LOG_DUMP_LIMIT {
				u.debug("logs: %s", u.logs)
			} else {
				u.debug("logs: appended=%d, ...%s", u.appended, u.logs[u.appended-LOG_DUMP_LIMIT:u.appended])
			}

		case <-ctx.Done():
			return
		}
	}
}

// Fetch fetches some logs within the range [begin, end).
// set end to -1 to fetch all logs till the end.
// it will block if no available log in the range unless NonBlocking is set in options.
func (u *LogKeeper) Fetch(begin, end int, options ...FetchOption) chan LogSlice {
	var flags FetchOption
	for _, v := range options {
		flags |= v
	}

	done := make(chan LogSlice, 1)
	u.fetchCh <- FetchEvent{begin, end, flags, done}
	return done
}

type FetchOption int

const (
	NonBlocking FetchOption = 1 << iota
)

func (u FetchOption) has(f FetchOption) bool {
	return u&f == f
}

type FetchEvent struct {
	begin   int
	end     int
	options FetchOption
	done    chan LogSlice
}

// AppendNew appends a new log and returns its index.
func (u *LogKeeper) AppendNew(command interface{}) int {
	done := make(chan int)
	u.appendCh <- AppendEvent{&LogEntry{Term: -1, Index: -1, Command: command}, done}
	return <-done
}

// Append appends a existing log.
func (u *LogKeeper) Append(log *LogEntry) int {
	done := make(chan int)
	u.appendCh <- AppendEvent{log, done}
	return <-done
}

type AppendEvent struct {
	log  *LogEntry
	done chan int
}

// UpdateTerm updates current term.
func (u *LogKeeper) UpdateTerm(term int) {
	done := make(chan bool)
	u.updateTermCh <- UpdateTermEvent{term, done}
	<-done
}

type UpdateTermEvent struct {
	term int
	done chan bool
}

// Trim removes all logs from begin to the last one.
// if there's no log at begin, Trim does nothing.
func (u *LogKeeper) Trim(begin int) {
	done := make(chan bool)
	u.trimCh <- TrimEvent{begin, done}
	<-done
}

type TrimEvent struct {
	begin int
	done  chan bool
}

// GetLog returns the log at index.
// if index is out of bound, returns nil.
func (u *LogKeeper) GetLog(index int) *LogEntry {
	done := make(chan *LogEntry)
	u.getLogCh <- GetLogEvent{index, done}
	return <-done
}

type GetLogEvent struct {
	index int
	done  chan *LogEntry
}

// GetLastLog returns the last log entry.
// if there's no log, returns nil.
func (u *LogKeeper) GetLastLog() *LogEntry {
	done := make(chan *LogEntry)
	u.getLastLogCh <- GetLastLogEvent{done}
	return <-done
}

type GetLastLogEvent struct {
	done chan *LogEntry
}

// Commit explicitly commits the log at index.
// if index exceeds the number of appended logs, the last log will be committed.
func (u *LogKeeper) Commit(index int) {
	done := make(chan bool)
	u.commitCh <- CommitEvent{index, done}
	<-done
}

type CommitEvent struct {
	index int
	done  chan bool
}

// GetCommitIndex returns the last committed log index.
// if no log has been committed, returns -1.
func (u *LogKeeper) GetCommitIndex() int {
	done := make(chan int)
	u.getCommitIndexCh <- GetCommitIndexEvent{done}
	return <-done
}

type GetCommitIndexEvent struct {
	done chan int
}

// Search searches for the last log satisfy pred.
// return -1 if no log satisfy pred.
func (u *LogKeeper) Search(pred func(int) bool) int {
	done := make(chan int)
	u.searchCh <- SearchEvent{pred, done}
	return <-done
}

type SearchEvent struct {
	pred func(int) bool
	done chan int
}

// Save writes all logs to the encoder.
func (u *LogKeeper) Save(e *labgob.LabEncoder) error {
	done := make(chan error)
	u.saveCh <- SaveEvent{e, done}
	return <-done
}

type SaveEvent struct {
	encoder *labgob.LabEncoder
	done    chan error
}

// Load reads logs from the decoder.
func (u *LogKeeper) Load(d *labgob.LabDecoder) error {
	done := make(chan error)
	u.loadCh <- LoadEvent{d, done}
	return <-done
}

type LoadEvent struct {
	decoder *labgob.LabDecoder
	done    chan error
}
